package com.lagou.common.zk;

import com.lagou.common.util.JSONSerializer;
import com.lagou.common.util.RpcService;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.zookeeper.CreateMode;

import java.util.*;


public class ServiceCenter {

    private static final String SERVICE_PATH = "/rpc-zookeeper/service";
    private static PathChildrenCache servicesCache;

    private ServiceCenter() {

    }

    public static void registerService(RpcService rpcService) throws Exception {

        String registerPath = SERVICE_PATH + "/" + rpcService.getId();
        JSONSerializer serializer = new JSONSerializer();
        byte[] data = serializer.serialize(rpcService);
        ZooKeeperHelper.createNode(registerPath, data, CreateMode.EPHEMERAL);

        System.out.println(rpcService.getId() + "已注册成功");
    }

    public static void updateServiceResponseTime(RpcService service, long responseTime) throws Exception {

        String nodeName = SERVICE_PATH + "/" + service.getId();
        byte[] data = ZooKeeperHelper.getData(nodeName);

        JSONSerializer serializer = new JSONSerializer();
        RpcService updatedService = serializer.deserialize(RpcService.class, data);
        updatedService.setResponseTime(responseTime);

        ZooKeeperHelper.setData(nodeName, serializer.serialize(updatedService));
    }

    public static RpcService nextService() throws Exception {

        if (servicesCache == null) {
            servicesCache = new PathChildrenCache(ZooKeeperHelper.getClient(), SERVICE_PATH, true);
            servicesCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        }

        // 1. Zookeeper记录每个服务端的最后一次响应时间，有效时间为5秒，5s内如果该服务端没有新的请求，响应时间清零或失效
        // 2. 当客户端发起调用，每次都选择最后一次响应时间短的服务端进行服务调用，如果时间一致，随机选取一个服务端进行调用，从而实现负载均衡
        List<RpcService> servicesBySort = new ArrayList<>();
        final long now = (new Date()).getTime();
        final JSONSerializer serializer = new JSONSerializer();

        servicesCache.getCurrentData().stream().forEach(e -> {
            try {
                long lastModifiedTime = e.getStat().getMtime();
                RpcService service = serializer.deserialize(RpcService.class, e.getData());
                if (now - lastModifiedTime > 5000) {
                    service.setResponseTime(0);
                }

                servicesBySort.add(service);

            } catch (Exception exception) {
                exception.printStackTrace();
            }
        });

        RpcService service = servicesBySort.stream().sorted(
                Comparator.comparing(RpcService::getResponseTime, (x, y) -> {
                            if (x > y) {
                                return 1;
                            } else if (x < y) {
                                return -1;
                            } else {
                                Random random = new Random();
                                return random.nextInt(3) - 1;
                            }
                        }
                )).findFirst().orElse(null);
        return service;
    }
}
